package com.yanggu.flink.datastream_api.async_io


import org.apache.commons.pool2.impl.GenericObjectPoolConfig
import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.async.{ResultFuture, RichAsyncFunction}
import redis.clients.jedis.JedisPool

import java.util.concurrent.CompletableFuture
import scala.concurrent.duration.SECONDS

/**
 * 异步IO常用于访问外部数据源, 用来丰富流数据。
 * <p>例如流中的数据是事实表, 去访问redis中的维表信息来进行维度染色
 * <p>需要注意的是异步io并不支持读写状态, 可以查看[[RichAsyncFunction]]的英文注释
 * <p>如果想要提高流应用的吞吐量可以使用如下策略：
 * <p>1. 可以使用本地缓存, 例如guava等进程内缓存框架来加速访问, 提高程序的响应速度
 * <p>2. 使用异步IO提高程序吞吐量
 * <p>3. 使用攒批的方式进行批读(例如MiniBatch)
 * <p><a href="https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/datastream/operators/asyncio/">flink官网异步io</a>
 *
 */
object AsyncIoDemo01 {

  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    environment.setParallelism(1)

    val source = environment.socketTextStream("localhost", 8888)

    val asyncDataStream = AsyncDataStream.orderedWait(source.keyBy(_ => true), new RedisAsyncFunction(), 60, SECONDS, 1000)

    asyncDataStream.print("async>>>>")

    environment.execute()

  }

}

class RedisAsyncFunction extends RichAsyncFunction[String, String] {

  private lazy val jedisPool = {
    new JedisPool(new GenericObjectPoolConfig(), "localhost", 6379, 3600/*, "Redis6379"*/)
  }

  //private lazy val valueState = getIterationRuntimeContext.getState(new ValueStateDescriptor[String]("value-state-test", classOf[String]))

  override def open(parameters: Configuration): Unit = {
    println("open方法被调用了")
  }

  override def asyncInvoke(input: String, resultFuture: ResultFuture[String]): Unit = {
    //println(valueState.value())
    //valueState.update(input)

    CompletableFuture.supplyAsync(() => {
      val resource = jedisPool.getResource
      val str = resource.get(input)
      resource.close()
      str
    }).whenComplete((v, ex) => {
      if (ex != null) {
        resultFuture.completeExceptionally(ex)
      } else {
        resultFuture.complete(Iterable(v))
      }
    })
  }

  override def timeout(input: String, resultFuture: ResultFuture[String]): Unit = {

  }

  override def close(): Unit = {
    println("close方法被调用了")
  }

}
